﻿using LJ.Abstractions.DependencyInjection;
using LJ.EventBus.HornedSerpent.Transport.Abstractions;
using LJ.EventBus.HornedSerpent.Transport.Abstractions.Exceptions;
using Microsoft.Extensions.DependencyInjection;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace LJ.EventBus.HornedSerpent.Transport.Kafka.Internal;


public class KafkaEventBusConsumerClientFactory(IDependencyInjectionFactory dependencyInjectionFactory) : IEventBusConsumerClientFactory
{
    private readonly IDependencyInjectionFactory _dependencyInjectionFactory = dependencyInjectionFactory;
    public IEventBusConsumerClient Create(string group, string topic)
    {
        try
        {
            using var serviceScope = _dependencyInjectionFactory.CreateScope();
            var serviceProvider = serviceScope.ServiceProvider;

            var consumerClient = serviceProvider.GetRequiredService<IEventBusConsumerClient>();

            if (consumerClient is not KafkaEventBusConsumerClient kafkaConsumerClient)
            {
                throw new ArgumentException($"{GetType().FullName} IConsumerClient is not {nameof(kafkaConsumerClient)}");
            }
            kafkaConsumerClient.Init(group, topic);
            return kafkaConsumerClient;

        }
        catch (Exception e)
        {
            throw new EventBusBrokerConnectionException(e);
        }
    }
}
